import common
import DCOracle2
import dispatcher
import string
import unittest

class test_DC(unittest.TestCase,dispatcher.Dispatcher):
    """unittests for DCOracle 2"""    

    def __init__(self,func,*args,**kw):
        unittest.TestCase.__init__(self,func)
        dispatcher.Dispatcher.__init__(self,func)


    def setUp(self):
        self.conn = DCOracle2.connect(common.getConnectionString())
        self.c    = self.conn.cursor()


    def tearDown(self):
        self.conn.close()
        self.conn=self.c=None


    def testConnect(self):
        """ try to connect to Oracle """
        pass


    def testCreateTable(self):
        """ try to create test table """

        try: self.c.execute("drop table partlist") 
        except: pass

        self.c.execute("create table partlist (part varchar2(62),  \
            partno integer, quantity integer, price number(10,2)) tablespace users")   

        file = open("namegen.out","r")

        v = []

        for l in file.readlines():
            (part, partno, quantity, price) = string.split(string.strip(l),',')
            v.append((part, partno, quantity, price))

        file.close()

        self.c.executemany("insert into partlist (part, partno, quantity, price) values (:1,:2,:3,:4)", v)
        self.conn.commit()


        self.c.execute("select part, partno, quantity, price from partlist order by partno")
        results = self.c.fetchall()


    def testConcurrentReads(self):
        self.dispatcher("testConcurrentReads",("funcConcurrentReads",3,(),{}))


    def funcConcurrentReads(self):

        env = self.th_setup()

        curs = self.conn.cursor()

        for i in range(5):
            curs.execute("select part, partno, quantity, price from partlist order by partno")
            results = curs.fetchall()

        self.th_teardown(env)


    def testCLOBS(self,iterations=30):
        """ testing CLOBs """
        self.funcLOBS(iterations,"CLOB")

    def testBLOBS(self,iterations=30):
        """ testing BLOBs """
        self.funcLOBS(iterations,"BLOB")


    def funcLOBS(self,iterations=30,t="CLOB"):
        """ test LOBS """

        try: self.c.execute("drop table testtab") 
        except: pass

        self.c.execute("create table testtab(num integer,content %s) tablespace users" % t)

        d={}
        for i in range(iterations):
            d[i] = chr(ord('a'))*100000


        for i in range(iterations):
            self.c.execute("insert into  testtab values(:1,EMPTY_%s())" % t , i)
            self.c.execute("select content from testtab where num=%d for update" % i)
            r = self.c.fetchone()
            lob = r[0]
            lob.write(d[i])
            self.conn.commit()


        self.c.execute("select num,content from testtab order by num")
        for r in self.c.fetchall():
            num = r[0]
            lob = r[1]
            data = lob.read()

            assert data==d[num],"Invalid data in LOB found"


    def testStoredProc1(self):
        """ testing describe() for stored procedures """

        db = self.conn

        d = db.describe('emp_actions')
        e = db.describe('emp')
        p = db.mapproc('emp_actions')


    def testStoredProc2(self):
        """ testing stored procedures """

        db      = self.conn

        find    = db.procedure.emp_actions.find
        findy   = db.procedure.emp_actions.findy

        self.c.execute('select empno, ename from emp')
        for r in self.c.fetchall():
            empid = r[0]
            ename = r[1]
            fe = find(empid)
            assert ename==fe,"Wrong employee name found %d vs. %d" % (ename,fe)
 

if __name__=="__main__":
    unittest.main()

        
